package setAPI.job;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
public class BatchDemoWithBroadcast {

	public static void main(String[] args) throws Exception {
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		DataSource<Tuple2<String, String>> toBroadcast = env.fromElements(
				Tuple2.of("env", "DEV"),
				Tuple2.of("vms_env", "QA")
		);

		DataSource<String> source = env.fromElements("A", "B", "C");
		MapOperator<String, String> end = source.map(new RichMapFunction<String, String>() {

			private Map<String, String> map = new HashMap<>();

			@Override
			public void open(Configuration parameters) throws Exception {
				super.open(parameters);
				List<Tuple2<String, String>> toBroadcastName = getRuntimeContext().getBroadcastVariable("toBroadcastName");
				for (Tuple2<String, String> tuple : toBroadcastName) {
					map.put(tuple.f0, tuple.f1);
				}
				log.warn("===>" + map);
			}

			@Override
			public String map(String s) throws Exception {
				return s;
			}
		}).withBroadcastSet(toBroadcast, "toBroadcastName");
		end.print();
	}
}
